package org.piggy.register.center.nacos;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingMaintainFactory;
import com.alibaba.nacos.api.naming.NamingMaintainService;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.Service;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.alibaba.nacos.common.utils.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.piggy.common.constants.BasicConst;
import org.piggy.common.constants.GatewayProtocol;
import org.piggy.common.invoker.BasicServiceInvoker;
import org.piggy.common.invoker.DubboServiceInvoker;
import org.piggy.common.invoker.ServiceInvoker;
import org.piggy.register.center.api.RegisterCenter;
import org.piggy.register.center.api.RegisterCenterHandler;
import org.piggy.common.config.ServiceDefinition;
import org.piggy.common.config.ServiceInstance;
import org.piggy.common.constants.GatewayConst;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;

@Slf4j
public class NacosRegisterCenter implements RegisterCenter {

    private String env;

    private NamingService namingService;

    private NamingMaintainService namingMaintainService;

    private final List<RegisterCenterHandler> registerCenterHandlerList = new CopyOnWriteArrayList<>();

    @Override
    public void init(String registerAddress, String env) {
        try {
            this.env = env;
            this.namingMaintainService = NamingMaintainFactory.createMaintainService(registerAddress);
            this.namingService = NamingFactory.createNamingService(registerAddress);
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void register(ServiceDefinition serviceDefinition, ServiceInstance serviceInstance) {
        try {
            Instance nacosInstance = new Instance();
            nacosInstance.setInstanceId(serviceInstance.getServiceInstanceId());
            nacosInstance.setPort(serviceInstance.getPort());
            nacosInstance.setIp(serviceInstance.getIp());
            Integer weight = serviceInstance.getWeight();
            if(weight == null){
                weight = 100;
            }
            nacosInstance.setWeight(weight);
            nacosInstance.setMetadata(Map.of(GatewayConst.META_DATA_WORD, JSONObject.toJSONString(serviceInstance)));

            namingService.registerInstance(serviceDefinition.getServiceName(), env, nacosInstance);
            // 0：表示无保护（默认值），仅返回健康实例。
            // 1：表示始终返回所有实例（无论健康状态）。
            namingMaintainService.updateService(serviceDefinition.getServiceName(), env, 0,
                    Map.of(GatewayConst.META_DATA_KEY, JSONObject.toJSONString(serviceDefinition)));

            log.info("register -> {} {}", serviceDefinition, serviceInstance);
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void deregister(ServiceDefinition serviceDefinition, ServiceInstance serviceInstance) {
        try {
            namingService.deregisterInstance(serviceDefinition.getServiceName(),
                    serviceDefinition.getEnvType(), serviceInstance.getIp(), serviceInstance.getPort());
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void subscribeAllServices(RegisterCenterHandler registerCenterHandler) {
        registerCenterHandlerList.add(registerCenterHandler);
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1, new NameThreadFactory("subscribe-services-pool"));
        scheduledThreadPool.scheduleWithFixedDelay(this::doSubscribeAllServices, 0, 10, TimeUnit.SECONDS);
    }

    private void doSubscribeAllServices() {
        try {
            //已经订阅的服务
            Set<String> subscribeService = namingService.getSubscribeServices().stream().map(ServiceInfo::getName).collect(Collectors.toSet());
            int pageNo = 1;
            int pageSize = 100;
            List<String> serviseList = namingService.getServicesOfServer(pageNo, pageSize, env).getData();

            while (CollectionUtils.isNotEmpty(serviseList)) {
                log.info("service list size {}", serviseList.size());

                for (String service : serviseList) {
                    if (subscribeService.contains(service)) {
                        continue;
                    }
                    EventListener eventListener = new NacosRegisterListener();
                    namingService.subscribe(service, env, eventListener);
                    log.info("subscribe {} {}", service, env);
                }
                serviseList = namingService.getServicesOfServer(++pageNo, pageSize, env).getData();
            }
        } catch (NacosException e) {
            e.printStackTrace();
        }
    }

    public class NacosRegisterListener implements EventListener {
        @Override
        public void onEvent(Event event) {
            if (event instanceof NamingEvent namingEvent) {
                String serviceName = namingEvent.getServiceName();
                try {
                    Service service = namingMaintainService.queryService(serviceName, env);

                    ServiceDefinition serviceDefinition = new ServiceDefinition();
                    serviceDefinition.setServiceName(service.getName());

                    Map<String, String> metadata = service.getMetadata();
                    String mateValue = metadata.get(GatewayConst.META_DATA_KEY);
                    JSONObject jsonObject = JSONObject.parseObject(mateValue);
                    if(jsonObject != null){
                        serviceDefinition.setServiceId(jsonObject.getString("serviceId"));
                        serviceDefinition.setVersion(jsonObject.getString("version"));
                        String protocol = jsonObject.getString(GatewayConst.PROTOCOL_KEY);
                        serviceDefinition.setProtocol(protocol);
                        serviceDefinition.setEnvType(jsonObject.getString("envType"));
                        serviceDefinition.setEnable(jsonObject.getBoolean("enable"));
                        serviceDefinition.setPatternPath(jsonObject.getString("patternPath"));
                        JSONObject invokerMap = jsonObject.getJSONObject("invokerMap");
                        Map<String, ServiceInvoker> newInvokerMap = new HashMap<>();
                        Set<Map.Entry<String, Object>> entries = invokerMap.entrySet();
                        for (Map.Entry<String, Object> entry : entries) {
                            String key = entry.getKey();
                            JSONObject value = (JSONObject) entry.getValue();
                            ServiceInvoker serviceInvoker;
                            if (GatewayProtocol.isDubbo(protocol)) {
                                serviceInvoker = value.to(DubboServiceInvoker.class);
                            } else {
                                serviceInvoker = value.to(BasicServiceInvoker.class);
                            }
                            newInvokerMap.put(key, serviceInvoker);
                        }
                        serviceDefinition.setInvokerMap(newInvokerMap);
                    }

                    List<Instance> allInstances = namingService.getAllInstances(service.getName(), env);
                    Set<ServiceInstance> set = new HashSet<>();
                    for (Instance instance : allInstances) {
                        ServiceInstance serviceInstance = new ServiceInstance();
                        serviceInstance.setServiceName(instance.getServiceName());
                        serviceInstance.setIp(instance.getIp());
                        serviceInstance.setPort(instance.getPort());
                        double weight = instance.getWeight();
                        serviceInstance.setWeight((int) weight);
                        serviceInstance.setServiceInstanceId(instance.getIp() + BasicConst.COLON_SEPARATOR + instance.getPort());
                        set.add(serviceInstance);
                    }
                    registerCenterHandlerList.forEach(new Consumer<RegisterCenterHandler>() {
                        @Override
                        public void accept(RegisterCenterHandler registerCenterHandler) {
                            registerCenterHandler.onChange(serviceDefinition, set);
                        }
                    });
                } catch (NacosException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
